package com.shujia.filter;

import com.shujia.been.Student;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class Code04RowFilter {
    static Connection connection;

    static {
        // 创建连接需要给定连接信息，所以需要给定配置类
        Configuration conf = new Configuration();

        // 由于对HBASE进行操作时，需要连接其Zookeeper 那么需要传入Zookeeper的IP
        conf.set("hbase.zookeeper.quorum", "node1,node2,master");
        // 获取连接对象 通过ConnectionFactory来创建连接对象

        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static boolean tableExists(Admin admin, TableName tableName) {
        boolean exists = false;
        try {
            exists = admin.tableExists(tableName);
            if (exists) {
                System.out.println("表存在..");
            } else {
                System.out.println("表不存在");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
        return exists;
    }


    // 需求：针对学生表，获取编号小于等于 1500100100 的所有学生
    public static void main(String[] args) throws IOException {

        Admin admin = connection.getAdmin();
        TableName tableName = TableName.valueOf("api:tb1");

        if (tableExists(admin, tableName)) {
            Table table = connection.getTable(tableName);

            Scan scan = new Scan();


            // 根据RowKey进行过滤数据，该方法相对速度较快，
            //   不需要加载全表中的数据进行做过滤,所以可以在表设计时，将经常使用的一些列放置RowKey中
            //
            RowFilter rowFilter = new RowFilter(
                    CompareFilter.CompareOp.LESS,
                    new BinaryComparator(Bytes.toBytes("1500100100!"))
            );

            scan.setFilter(rowFilter);

            ResultScanner scanner = table.getScanner(scan);

            printData(scanner);

            admin.close();
            table.close();
            connection.close();

        }


    }

    public static void printData(ResultScanner scanner) {
        for (Result result : scanner) {
            String rowKey = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
                    , "name".getBytes(StandardCharsets.UTF_8)));

            int age = Bytes.toInt(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
                    , "age".getBytes(StandardCharsets.UTF_8)));

            String gender = Bytes.toString(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
                    , "gender".getBytes(StandardCharsets.UTF_8)));

            String clazz = Bytes.toString(result.getValue("info2".getBytes(StandardCharsets.UTF_8)
                    , "clazz".getBytes(StandardCharsets.UTF_8)));

            Student student = new Student(rowKey, name, age, gender, clazz);
            System.out.println(student);

//            System.out.println("name:"+name+" age:"+age+" gender:"+gender+" clazz:"+clazz);

        }
    }
}
